package io.github.wanggit.antrpc.client.monitor.report;

import io.github.wanggit.antrpc.commons.bean.IdGenHelper;
import io.github.wanggit.antrpc.commons.test.ForTest;
import io.github.wanggit.antrpc.commons.utils.LongToDateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import zipkin2.Endpoint;
import zipkin2.Span;
import zipkin2.codec.SpanBytesEncoder;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class KafkaTracingSender implements ITracingSender, Runnable {

    private static KafkaTracingSender instance = null;

    /** 存入方法跟踪信息的队列 */
    private static final ConcurrentLinkedQueue<ZipkinAllMethodTracing.TracingObj> tracingObjQueue =
            new ConcurrentLinkedQueue<>();

    /** 定时把方法跟踪信息发送到Kafka的定时任务线程池 */
    private static final ScheduledThreadPoolExecutor scheduledExecutor =
            new ScheduledThreadPoolExecutor(1);

    /** 当跟踪信息太多 scheduledExecutor 处理不过来时，当队列中数量超过 NEED_TRACING_HELP 时，启动此线程池帮助处理 */
    private static final ThreadPoolExecutor busyHelperThreadPool =
            new ThreadPoolExecutor(
                    1,
                    3,
                    100,
                    TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(10),
                    (r, executor) -> {
                        if (log.isWarnEnabled()) {
                            log.warn("Tracing Object too many..");
                        }
                    });

    /** 定时任务线程池处理时队列剩余的最大值，超过这个值，就会使用 busyHelperThreadPool 帮忙处理。 */
    private static final int NEED_TRACING_HELP = 1000;

    private static final int BATCH_SIZE = 300;

    private static final Builder builder = new Builder();

    private KafkaTemplate kafkaTemplate;

    private String topic;

    private Endpoint localEndpoint;

    private boolean reportAllMethodsToZipkin;

    private final AtomicInteger counter = new AtomicInteger(0);

    private KafkaTracingSender() {
        scheduledExecutor.scheduleAtFixedRate(this, 20, 150, TimeUnit.MILLISECONDS);
    }

    @Override
    public void run() {
        trace(fillSpanList(BATCH_SIZE));
        while (isBusy()) {
            busyHelperThreadPool.submit(() -> trace(fillSpanList(BATCH_SIZE * 2)));
        }
    }

    boolean isBusy() {
        return counter.get() > NEED_TRACING_HELP;
    }

    @ForTest
    int getCounterValue() {
        return counter.get();
    }

    List<Span> fillSpanList(int cnt) {
        List<Span> spans = new ArrayList<>();
        for (int i = 0; i < cnt; i++) {
            ZipkinAllMethodTracing.TracingObj tracingObj = tracingObjQueue.poll();
            if (null != tracingObj) {
                counter.decrementAndGet();
                spans.add(createSpan(tracingObj));
            }
        }
        return spans;
    }

    private Span createSpan(ZipkinAllMethodTracing.TracingObj tracingObj) {
        String methodName = convertMethodName(tracingObj.getMethodName());
        String lowHex16ParentId =
                IdGenHelper.to16LowHexZipkinId(tracingObj.getParent().getRequestId());
        String lowHex16Id = IdGenHelper.to16LowHexZipkinId(tracingObj.getCurrent().getRequestId());
        return Span.newBuilder()
                .timestamp(tracingObj.getBegin())
                .parentId(lowHex16ParentId)
                .id(lowHex16Id)
                .traceId(tracingObj.getCurrent().getSerialNumber())
                .name(methodName)
                .localEndpoint(localEndpoint)
                .remoteEndpoint(localEndpoint)
                .kind(
                        IdGenHelper.isZipkinIdAllZero(lowHex16ParentId)
                                ? Span.Kind.CLIENT
                                : Span.Kind.SERVER)
                .duration(tracingObj.getEnd() - tracingObj.getBegin())
                .putTag("0_className", convertClassName(tracingObj.getClassName()))
                .putTag("1_methodName", methodName)
                .putTag("2_start", LongToDateUtil.toDateMsStr(tracingObj.getBegin() / 1000, ""))
                .putTag("3_end", LongToDateUtil.toDateMsStr(tracingObj.getEnd() / 1000, ""))
                .putTag(
                        "4_rt(ms)",
                        String.valueOf((tracingObj.getEnd() - tracingObj.getBegin()) / 1000))
                .putTag("5_requestId", tracingObj.getCurrent().getRequestId())
                .putTag("6_parentId", tracingObj.getParent().getRequestId())
                .build();
    }

    public void addTracingObj(ZipkinAllMethodTracing.TracingObj tracingObj) {
        tracingObjQueue.offer(tracingObj);
        counter.incrementAndGet();
    }

    private static String convertClassName(String className) {
        return className.replace("/", ".");
    }

    static String convertMethodName(String longMethodName) {
        int idx = longMethodName.indexOf("(");
        int dotIdx = longMethodName.lastIndexOf(".", idx);
        return longMethodName.substring(dotIdx + 1);
    }

    public static Builder builder() {
        return builder;
    }

    public static KafkaTracingSender getInstance() {
        return instance;
    }

    public boolean isReportAllMethodsToZipkin() {
        return reportAllMethodsToZipkin;
    }

    @Override
    public void trace(List<Span> spans) {
        if (null != spans && !spans.isEmpty() && null != kafkaTemplate) {
            if (log.isDebugEnabled()) {
                log.debug("Zipkin Tracing Spans Size: " + spans.size());
            }
            byte[] value = SpanBytesEncoder.JSON_V2.encodeList(spans);
            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(topic, value);
            kafkaTemplate.send(record);
        }
    }

    static class Builder {

        private Builder() {}

        private String topic;

        private KafkaTemplate kafkaTemplate;

        private boolean reportAllMethodsToZipkin;

        private Endpoint localEndpoint;

        private final AtomicBoolean builded = new AtomicBoolean(false);

        public Builder setLocalEndpoint(Endpoint localEndpoint) {
            this.localEndpoint = localEndpoint;
            return this;
        }

        public Builder setTopic(String topic) {
            this.topic = topic;
            return this;
        }

        public Builder setKafkaTemplate(KafkaTemplate kafkaTemplate) {
            this.kafkaTemplate = kafkaTemplate;
            return this;
        }

        public Builder setReportAllMethodsToZipkin(boolean reportAllMethodsToZipkin) {
            this.reportAllMethodsToZipkin = reportAllMethodsToZipkin;
            return this;
        }

        public synchronized KafkaTracingSender build() {
            if (!builded.compareAndSet(false, true)) {
                throw new UnsupportedOperationException();
            }
            instance = new KafkaTracingSender();
            instance.topic = topic;
            instance.kafkaTemplate = kafkaTemplate;
            instance.reportAllMethodsToZipkin = reportAllMethodsToZipkin;
            instance.localEndpoint = localEndpoint;
            return instance;
        }
    }
}
